package main

import (
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/latifrons/gows"
	"github.com/sirupsen/logrus"
	"github.com/spf13/viper"
	"net/http"
	"ogbrowser-push-service/service"
	"time"
)

func handleFunc(client *gows.Client, bytes []byte) {
	fmt.Println(client.Conn.RemoteAddr().String())
	fmt.Println(string(bytes))
}

func main() {
	viper.AutomaticEnv()

	r := gin.New()
	r.Use(gows.GinLogger(logrus.StandardLogger(), time.RFC3339, true))

	hub := gows.NewHub()
	go hub.Run(false)

	r.GET("/ws", func(c *gin.Context) {
		c.Header("Access-Control-Allow-Origin", "*")
		gows.ServeWs(hub, c.Writer, c.Request, handleFunc)
	})

	r.GET("/health", func(c *gin.Context) {
		c.JSON(http.StatusOK, gin.H{
			"status": "ok",
		})
	})

	go readKafka(hub)
	r.Run("0.0.0.0:8080")
}

func readKafka(hub *gows.Hub) {
	brokers := []string{viper.GetString("kafka.broker")}
	topic := viper.GetString("kafka.topic")

	res := make(chan string)
	//groupId := "og-visual-test"

	go service.KafkaConsumer(brokers, topic, res)

	for {
		select {
		case msg := <-res:
			fmt.Println(msg)
			hub.Broadcast([]byte(msg))
		}
	}
}
